package com.hyjr.RxTask.rabbitmq;

import java.io.IOException;

import com.hyjr.RxTask.work.service.ConsumerWork;
import com.hyjr.common.utils.LogUtils;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

/**
 * @program: rabbit-learn
 * @description: 消费者
 * @author: 535504
 * @create: 2018-04-26 15:32
 **/
public class ConsumerUtil {
	
	public static void consumer(String queueName, ConsumerWork consumerWork) throws Exception {
		LogUtils.debug("开启队列消费，队列名称：" + queueName);
		Connection connection = ConnectionUtil.getConnection();
		Channel channel = connection.createChannel();
		channel.queueDeclare(queueName, false, false, false, null);
		channel.basicQos(1);
		// 自4.0+ 版本后无法再使用QueueingConsumer，而官方推荐使用DefaultConsumer
		com.rabbitmq.client.Consumer consumer = new DefaultConsumer(channel) {
			@Override
			public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
					throws IOException {
				super.handleDelivery(consumerTag, envelope, properties, body);
				try {
					String message = new String(body, "UTF-8");
					
					consumerWork.doWork(message);
					
					channel.basicAck(envelope.getDeliveryTag(), false);
				} catch (Exception e) {
					channel.basicAck(envelope.getDeliveryTag(), true);
					LogUtils.debug("检测端，ConsumerUtil，执行失败，队列名：" + queueName, e);
				}
			}
		};
		// 监听队列，当b为true时，为自动提交（只要消息从队列中获取，无论消费者获取到消息后是否成功消息，都认为是消息已经成功消费），
		// 当b为false时，为手动提交（消费者从队列中获取消息后，服务器会将该消息标记为不可用状态，等待消费者的反馈，
		// 如果消费者一直没有反馈，那么该消息将一直处于不可用状态。
		// 如果选用自动确认,在消费者拿走消息执行过程中出现宕机时,消息可能就会丢失！！）
		// 使用channel.basicAck(envelope.getDeliveryTag(),false);进行消息确认
		channel.basicConsume(queueName, false, consumer);
		LogUtils.debug("队列消费结束，队列名称：" + queueName);
	}
}
